#ifndef __M_CHANNEL_H__
#define __M_CHANNEL_H__

#include "muduo/proto/codec.h"
#include "muduo/proto/dispatcher.h"
#include "muduo/net/TcpConnection.h"
#include "../mqcommon/logger.hpp"
#include "../mqcommon/helpper.hpp"
#include "../mqcommon/msg.pb.h"
#include "../mqcommon/mq_proto.pb.h"
#include "../mqcommon/threadpool.hpp"

#include "consumer.hpp"
#include <mutex>
#include <condition_variable>
#include <unordered_map>

namespace mmq
{
    using MessagePtr = std::shared_ptr<google::protobuf::Message>;
    using ProtobufCodecPtr = std::shared_ptr<ProtobufCodec>;
    using BasicCommonResponsePtr = std::shared_ptr<BasicCommonResponse>;
    using BasicConsumeResponsePtr = std::shared_ptr<BasicConsumeResponse>;
    class Channel
    {
    public:
        using ptr = std::shared_ptr<Channel>;
    public:
        Channel(const muduo::net::TcpConnectionPtr &conn,
            const ProtobufCodecPtr &codec):
            _cid(UuidHelper::uuid()),
            _conn(conn),
            _codec(codec)
        {}
        ~Channel()
        {
            BasicCancel();
        }
        std::string Cid()
        {
            return _cid;
        }

        bool OpenChannel()
        {
            OpenChannelRequest req;
            std::string rid = UuidHelper::uuid();
            req.set_rid(rid);
            req.set_cid(_cid);
            _codec->send(_conn, req);
            BasicCommonResponsePtr resp = WaitResponse(rid);
            return resp->ok();
        }
        void CloseChannel()
        {
            CloseChannelRequest req;
            std::string rid = UuidHelper::uuid();
            req.set_rid(rid);
            req.set_cid(_cid);
            _codec->send(_conn, req);
            WaitResponse(rid);
        }

        bool DeclareExchange(const std::string &name, mmq::ExchangeType type, bool durable,
                bool auto_delete, google::protobuf::Map<std::string, std::string> &args)
        {
            // 构造一个声明虚拟机的请求对象
            DeclareExchangeRequest req;
            std::string rid = UuidHelper::uuid();
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_exchange_name(name);
            req.set_exchange_type(type);
            req.set_durable(durable);
            req.set_auto_delete(auto_delete);
            req.mutable_args()->swap(args);
            // 向服务器发送请求
            _codec->send(_conn, req);
            // 等待服务器响应
            BasicCommonResponsePtr resp = WaitResponse(rid);
            return resp->ok();
        }
        void DeleteExchange(const std::string &ename)
        {
            DeclareExchangeRequest req;
            std::string rid = UuidHelper::uuid();
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_exchange_name(ename);
            WaitResponse(rid);
        }

        bool DeclareQueue(const std::string &name, bool durable, bool exclusive, bool auto_delete, 
            google::protobuf::Map<std::string, std::string> &args)
        {
            DeclareQueueRequest req;
            std::string rid = UuidHelper::uuid();
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_queue_name(name);
            req.set_durable(durable);
            req.set_exclusive(exclusive);
            req.set_auto_delete(auto_delete);
            req.mutable_args()->swap(args);
            _codec->send(_conn, req);
            BasicCommonResponsePtr resp = WaitResponse(rid);
            return resp->ok();
        }
        void DeleteQueue(const std::string &qname)
        {
            DeleteQueueRequest req;
            std::string rid = UuidHelper::uuid();
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_queue_name(qname);
            _codec->send(_conn, req);
            WaitResponse(rid);
        }

        bool QueueBind(const std::string &ename, const std::string &qname, const std::string &key)
        {
            QueueBindRequest req;
            std::string rid = UuidHelper::uuid();
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_exchange_name(ename);
            req.set_queue_name(qname);
            req.set_binding_key(key);
            _codec->send(_conn, req);
            BasicCommonResponsePtr resp = WaitResponse(rid);
            return resp->ok();
        }
        void QueueUnBind(const std::string &ename, const std::string &qname)
        {            
            QueueUnBindRequest req;
            std::string rid = UuidHelper::uuid();
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_exchange_name(ename);
            req.set_queue_name(qname);
            _codec->send(_conn, req);
            WaitResponse(rid);
        }

        void BasicPublish(const std::string &ename, BasicProperties *bp, const std::string &body)
        {
            BasicPublishRequest req;
            std::string rid = UuidHelper::uuid();
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_exchange_name(ename);
            req.set_body(body);
            if(bp)
            {
                req.mutable_properties()->set_id(bp->id());
                req.mutable_properties()->set_delivery_mode(bp->delivery_mode());
                req.mutable_properties()->set_routing_key(bp->routing_key());
            }
            _codec->send(_conn, req);
            WaitResponse(rid);
        }
        void BasicAck(const std::string &msg_id)
        {
            if(_consumer.get() == nullptr)
            {
                DLOG("消息确认时，找不到消费者信息！");
                return;
            }
            BasicAckRequest req;
            std::string rid = UuidHelper::uuid();
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_queue_name(_consumer->_qname);
            req.set_message_id(msg_id);
            _codec->send(_conn, req);
            WaitResponse(rid);
        }

        bool BasicConsume(const std::string &qname, const std::string &consumer_tag, bool ack_flag, 
            const ConsumerCallBack &cb)
        {
            if(_consumer.get())
            {
                DLOG("当前信道已订阅其他队列消息！");
                return false;
            }
            BasicConsumeRequest req;
            std::string rid = UuidHelper::uuid();
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_queue_name(qname);
            req.set_consumer_tag(consumer_tag);
            req.set_auto_ack(ack_flag);
            _codec->send(_conn, req);
            BasicCommonResponsePtr resp = WaitResponse(rid);
            if(resp->ok() == false)
            {
                DLOG("订阅失败！");
                return false;
            }
            _consumer = std::make_shared<Consumer>(consumer_tag, qname, ack_flag, cb);
            return true;
        }
        void BasicCancel()
        {
            if(_consumer.get() == nullptr)
            {
                // DLOG("取消订阅时，找不到订阅者信息");
                return;
            }
            BasicCancelRequest req;
            std::string rid = UuidHelper::uuid();
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_queue_name(_consumer->_qname);
            req.set_consumer_tag(_consumer->_tag);
            _codec->send(_conn, req);
            WaitResponse(rid);
            _consumer.reset();
        }
    public:
        // 连接收到基础响应后，向hash_map中添加响应
        void PutBasicResponse(const BasicCommonResponsePtr &resp)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _basic_resp.insert({resp->rid(), resp});
            _cv.notify_all();
        }
        // 连接收到消息推送后，需要通过信道找到对应的消费者对象，通过回调函数进行消息处理
        void Consume(const BasicConsumeResponsePtr &resp)
        {
            if(_consumer.get() == nullptr)
            {
                DLOG("消息处理时，未找到订阅者信息！");
                return;
            }
            if(_consumer->_tag != resp->consumer_tag())
            {
                DLOG("收到的推送消息中的消费者标识，与当前信道消费者标识不一致！");
                return;
            }
            _consumer->_callback(resp->consumer_tag(), resp->mutable_properties(), resp->body());
        }
    private:
        BasicCommonResponsePtr WaitResponse(const std::string &rid)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _cv.wait(lock, [&rid, this](){
                return _basic_resp.find(rid) != _basic_resp.end();
            });
            BasicCommonResponsePtr resp = _basic_resp[rid];
            _basic_resp.erase(rid);
            return resp;
        }
    private:
        std::string _cid;
        muduo::net::TcpConnectionPtr _conn;
        ProtobufCodecPtr _codec;
        Consumer::ptr _consumer;
        std::mutex _mutex;
        std::condition_variable _cv;
        std::unordered_map<std::string, BasicCommonResponsePtr> _basic_resp;
    };

    class ChannelManager
    {
    public:
        using ptr = std::shared_ptr<ChannelManager>;
    public:
        ChannelManager()
        {}
        Channel::ptr Create(const muduo::net::TcpConnectionPtr &conn, const ProtobufCodecPtr &codec)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            Channel::ptr channel = std::make_shared<Channel>(conn, codec);
            _channels.insert({channel->Cid(), channel});
            return channel;
        }
        void Remove(const std::string &cid)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _channels.erase(cid);
        }
        Channel::ptr Get(const std::string &cid)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            auto it = _channels.find(cid);
            if(it == _channels.end())
            {
                return Channel::ptr();
            }
            return it->second;
        }
    private:
        std::mutex _mutex;
        std::unordered_map<std::string, Channel::ptr> _channels; 
    };

}

#endif